package timo.rabbitmq.test.study.topic;

import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import timo.rabbitmq.common.CommonUtils;

import java.io.IOException;
import java.util.Scanner;

/**
 * 拆分客户端是因为main方法才能输入 new Scanner(System.in).nextLine().split("\\s");
 *
 * 客户端启动前已发布的消息接收不了
 *
 * @author timo
 * @date 2022/7/11
 */
public class TopicTestClient {
    @SneakyThrows
    public static void main(String[] args) {
        Connection connection = CommonUtils.getConnection();
        Channel channel = connection.createChannel();

        //一次只接收一条消息
        channel.basicQos(1);
        channel.exchangeDeclare("topic_logs", BuiltinExchangeType.TOPIC);
        //自动生成队列名,
        //非持久,独占,自动删除
        String queueName = channel.queueDeclare().getQueue();
        System.out.println("输入bindingKey,用空格隔开:");
        //例子 ：lazy.# *.orange.* *.*.rabbit
        String[] a = new Scanner(System.in).nextLine().split("\\s");
        //把该队列,绑定到 topic_logs 交换机
        //允许使用多个 bindingKey
        for (String bindingKey  : a) {
            channel.queueBind(queueName, "topic_logs", bindingKey );
        }
        System.out.println("等待接收数据");

        //收到消息后用来处理消息的回调对象
        DeliverCallback callback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                String msg = new String(message.getBody(), "UTF-8");
                System.out.println("收到: "+msg);

                //遍历字符串中的字符,每个点使进程暂停一秒
                for (int i = 0; i < msg.length(); i++) {
                    if (msg.charAt(i)=='.') {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                        }
                    }
                }
                System.out.println("处理结束");
                //autoAck设置为false 手动发送回执
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        //消费者取消时的回调对象
        CancelCallback cancel = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
            }
        };
        //autoAck设置为false,则需要手动确认发送回执
        //这里为true的情况下，会出现消息丢失，
        // 如这个客户端处理1.......花费时间过长，其他轮询分配到的任务没有处理，就直接挂掉了
        // 例：发送 1...... 2 3 4 5  启动两个客户端 消息35 未被处理
//        channel.basicConsume("helloworld", true, callback, cancel);
        //需要手动确认发送回执 不然一直不会被消费
        //channel.basicConsume("helloworld", false, callback, cancel);
        channel.basicConsume(queueName, false, callback, cancel);

        Thread.sleep(100000);
    }

}
